package com.lecosa.flink;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;

//mysql 开这3个
// server_id=2
//log_bin=mysql-bin
//binlog_format=ROW

public class FlinkCDC {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		// 创建 CDC Source
		DebeziumSourceFunction<String> mySqlSource = createCDCSource();

		// 添加 CDC Source 到环境中
		DataStreamSource<String> dataStreamSource = env.addSource(mySqlSource);

		// 打印输出
		dataStreamSource.print();

		// 执行任务
		env.execute();
	}

	// 创建 CDC Source 方法
	public static DebeziumSourceFunction<String> createCDCSource() {
		return MySqlSource.<String>builder()
				.hostname("110.41.6.107")
				.port(33306)
				.username("root")
				.password("root")
				.databaseList("scm_kaifa")
				.tableList("scm_kaifa.scf_payment_info")
				.deserializer(new StringDebeziumDeserializationSchema())
				.startupOptions(StartupOptions.initial())
				.build();
	}
//	  程序在停掉之前要做 savepoint   启动的时候要加上checkpoint的地址。。  断点续传的功能。。
	
	
	/***
	一次只能读一张表 ,不需要序列化 
	
	flinksql cdc  要在flink 1.13版本之上 
	
	flinksql 只能是单表

***/
}